package com.navinfo.tripanalysis.offline.accumulator;

import com.navinfo.tripanalysis.offline.util.AccumulatorUtils;
import org.apache.spark.util.AccumulatorV2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 0200异常数据累加器，统计标准异常指标
 * @author hmc
 */

public class Load0200DataAccumulator extends AccumulatorV2<String,String> {

    private static final Logger logger = LoggerFactory.getLogger(Load0200DataAccumulator.class);

    //设置默认累加值
    private static final int ACCUMULATOR_VALUE = 1;

    //设置各统计指标原始状态
    private String zero = DCConstants.ACC_0200_COUNT+"=0|"
                            +DCConstants.ACC_0200_REPEATED_0_10_Ratio+"=0|"
                            +DCConstants.ACC_0200_REPEATED_10_50_Ratio+"=0|"
                            +DCConstants.ACC_0200_REPEATED_50_Ratio+"=0|"
                            +DCConstants.ACC_0200_MILEAGE_10_100_COUNT+"=0|"
                            +DCConstants.ACC_0200_MILEAGE_100_300_COUNT+"=0|"
                            +DCConstants.ACC_0200_MILEAGE_300_COUNT+"=0|"
                            +DCConstants.ACC_0200_FUEL_10_100_COUNT+"=0|"
                            +DCConstants.ACC_0200_FUEL_100_300_COUNT+"=0|"
                            +DCConstants.ACC_0200_FUEL_300_COUNT+"=0|"
                            +DCConstants.ACC_0200_LAT_LON_10_100_COUNT+"=0|"
                            +DCConstants.ACC_0200_LAT_LON_100_300_COUNT+"=0|"
                            +DCConstants.ACC_0200_LAT_LON_300_COUNT+"=0"
                            ;

    //初始化结果
    private String data = zero;

    @Override
    public boolean isZero() {
        return data.equalsIgnoreCase(zero);
    }

    @Override
    public AccumulatorV2<String, String> copy() {
        return new Load0200DataAccumulator();
    }

    @Override
    public void reset() {
        data = zero;
    }

    @Override
    public void add(String v) {
        data = AccumulatorUtils.mergeData(data,"\\|",v,ACCUMULATOR_VALUE);
    }

    @Override
    public void merge(AccumulatorV2<String, String> other) {
        data = AccumulatorUtils.mergeAccumulator(data,"\\|",other);
    }

    @Override
    public String value() {
        return data;
    }
}
